package cc.csln.so.cdelayqueue.jvm;

import cc.csln.so.cdelayqueue.CDelayQueueConfig;
import cc.csln.so.cdelayqueue.DelayedTask;
import cc.csln.so.cdelayqueue.IDelayQueue;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.function.BiConsumer;

/**
 * JvmDelayQueue
 * <p>
 * 基于 JDK {@link DelayQueue} 的 JVM 内存型延迟队列实现：
 * - 通过 key 去重，新增同 key 任务会刷新触发时间与负载
 * - 支持任务级回调，消费线程为守护线程，自动在应用启动后运行
 * - 适用于单机场景，进程内存存储，不保证跨进程/重启后的任务持久性
 *
 * @author Cusalt
 * @since 2025/10/16 17:31
 **/
@Slf4j
@Component
@RequiredArgsConstructor
public class JvmDelayQueue implements IDelayQueue, InitializingBean {

    /**
     * 延迟队列，存储待触发的任务。
     */
    private final DelayQueue<DelayedTask> queue = new DelayQueue<>();

    /**
     * 任务映射，用于按 key 去重与刷新任务。
     */
    private final Map<String, DelayedTask> taskMap = new ConcurrentHashMap<>();

    /**
     * JVM 延迟队列配置（来自 Spring 配置绑定）。
     */
    private final CDelayQueueConfig.JvmDelayQueueConfig config;

    /**
     * 回调执行线程池。
     * 来源：自动装配的 `ThreadPoolTaskExecutor`。
     * 用途：在守护线程中取到任务后，异步执行任务回调，避免阻塞队列消费。
     */
    private final ThreadPoolTaskExecutor executor;

    /**
     * 获取默认延迟毫秒数。
     * 来源：`c-delay-queue.jvm.delay-ms`。
     */
    @Override
    public long getDefaultDelayMs() {
        return config.getDelayMs();
    }

    /**
     * Bean 初始化完成后回调。
     * 行为：启动守护消费者线程。
     * 说明：替代 `@PostConstruct`，避免容器初始化阶段的顺序问题。
     */
    @Override
    public void afterPropertiesSet() {
        startConsumer();
    }

    /**
     * 新增或刷新延迟任务（按 key 去重），支持任务级回调。
     * <p>
     * 逻辑：
     * - 当 key 已存在时：从队列移除旧任务，刷新触发时间、负载与回调，再重新入队
     * - 当 key 不存在时：创建新任务、记录映射并入队
     * <p>
     * 线程安全：使用 synchronized(this) 保证映射与队列操作的一致性。
     *
     * @param key      任务 key，不可为空或空白
     * @param payload  任务负载，可为空
     * @param delayMs  延迟毫秒，>=0
     * @param callback 任务触发时调用，不能为空
     */
    @Override
    public void addTask(String key, String payload, long delayMs, BiConsumer<String, String> callback) {
        if (key == null || key.trim().isEmpty()) {
            throw new IllegalArgumentException("key不能为空");
        }
        if (callback == null) {
            throw new IllegalArgumentException("callback不能为空");
        }
        long newTriggerTime = System.currentTimeMillis() + delayMs;
        synchronized (this) {
            DelayedTask existing = taskMap.get(key);
            if (existing != null) {
                // 从队列移除旧任务，刷新触发时间与内容
                queue.remove(existing);
                existing.setPayload(payload);
                existing.setTriggerTime(newTriggerTime);
                existing.setCallback(callback);
                queue.put(existing);
                log.debug("刷新延迟任务: key={}, delayMs={}, count={}", key, delayMs, queue.size());
            } else {
                DelayedTask task = new DelayedTask(key, payload, newTriggerTime, callback);
                taskMap.put(key, task);
                queue.put(task);
                log.debug("新增延迟任务: key={}, delayMs={}, count={}", key, delayMs, queue.size());
            }
        }
    }

    /**
     * 启动延迟任务消费者线程（守护线程）。
     * <p>
     * 线程行为：阻塞获取队列头任务，取出即从映射移除；若存在回调则执行并捕获异常。
     */
    public void startConsumer() {
        Thread t = new Thread(() -> {
            while (!Thread.interrupted()) {
                try {
                    DelayedTask task = queue.take();
                    // 取出后移除映射
                    taskMap.remove(task.getKey());

                    BiConsumer<String, String> cb = task.getCallback();
                    if (cb != null) {
                        try {
                            log.debug("延迟任务消费: key={}, payload={}, count={}", task.getKey(), task.getPayload(), queue.size());
                            executor.execute(() -> {
                                long start = System.currentTimeMillis();
                                log.debug("延迟任务回调开始: key={}, payload={}", task.getKey(), task.getPayload());
                                cb.accept(task.getKey(), task.getPayload());
                                log.debug("延迟任务回调完成: key={}, cost={}ms", task.getKey(), System.currentTimeMillis() - start);
                            });
                        } catch (Exception ex) {
                            log.error("延迟任务执行异常: key={}, payload={}, count={}", task.getKey(), task.getPayload(), queue.size(), ex);
                        }
                    } else {
                        log.debug("延迟任务触发: key={}, payload={}, count={}", task.getKey(), task.getPayload(), queue.size());
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "c-dq-daemon");
        t.setDaemon(true);
        t.start();
    }

}
